package org.codehaus.activemq.message;

import java.io.DataInput;
import java.io.IOException;
import org.codehaus.activemq.util.BitArray;

public class ActiveMQMessageReader extends AbstractPacketReader
{
  public int getPacketType()
  {
    return 6;
  }

  public Packet createPacket()
  {
    return new ActiveMQMessage();
  }

  public void buildPacket(Packet packet, DataInput dataIn)
    throws IOException
  {
    ActiveMQMessage msg = (ActiveMQMessage)packet;
    msg.setJMSMessageID(super.readUTF(dataIn));
    msg.setJMSClientID(super.readUTF(dataIn));
    msg.setProducerID(super.readUTF(dataIn));
    msg.setJMSDestination(ActiveMQDestination.readFromStream(dataIn));
    msg.setJMSDeliveryMode(dataIn.readByte());
    msg.setJMSPriority(dataIn.readByte());
    BitArray ba = new BitArray();
    ba.readFromStream(dataIn);
    msg.setJMSRedelivered(ba.get(8));
    msg.setXaTransacted(ba.get(9));

    if (ba.get(0)) {
      msg.setJMSCorrelationID(super.readUTF(dataIn));
    }
    if (ba.get(1)) {
      msg.setJMSType(super.readUTF(dataIn));
    }
    if (ba.get(2)) {
      msg.setEntryBrokerName(super.readUTF(dataIn));
    }
    if (ba.get(3)) {
      msg.setEntryClusterName(super.readUTF(dataIn));
    }
    if (ba.get(4)) {
      msg.setTransactionId(super.readUTF(dataIn));
    }
    if (ba.get(5)) {
      msg.setJMSReplyTo(ActiveMQDestination.readFromStream(dataIn));
    }
    if (ba.get(6)) {
      msg.setJMSTimestamp(dataIn.readLong());
    }
    if (ba.get(7)) {
      msg.setJMSExpiration(dataIn.readLong());
    }
    if (ba.get(10)) {
      int cidlength = dataIn.readShort();
      if (cidlength > 0) {
        int[] cids = new int[cidlength];
        for (int i = 0; i < cids.length; i++) {
          cids[i] = dataIn.readShort();
        }
        msg.setConsumerNos(cids);
      }
    }
    if (ba.get(11)) {
      msg.setProperties(msg.readMapProperties(dataIn));
    }
    if (ba.get(12)) {
      int payloadLength = dataIn.readInt();
      if (payloadLength >= 0) {
        byte[] payload = new byte[payloadLength];
        dataIn.readFully(payload);
        msg.setBodyAsBytes(payload);
      }
    }
  }
}